package com.atguigu.flink.chapter11;

import com.atguigu.flink.chapter05.Source.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/1/27 10:11
 */
public class Flink08_SQL_DDL {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 使用sql的DDL语法，连接kafka，作为source
        tableEnv.executeSql("create table source_sensor (id string, ts bigint, vc int) with (" + "'connector' = 'kafka',"
                + "'topic' = 'topic_source_sensor',"
                + "'properties.bootstrap.servers' = 'hadoop1:9092,hadoop2:9092,hadoop3:9092',"
                + "'properties.group.id' = 'atguigu',"
                + "'scan.startup.mode' = 'latest-offset',"
//                + "'format' = 'json'"
                + "'format' = 'csv'"
                + ")");


        // 使用sql的DDL语法，连接kafka，作为sink
        tableEnv.executeSql("create table sink_sensor (id string, ts bigint, vc int) with ("
                + "'connector' = 'kafka',"
                + "'topic' = 'topic_sink_sensor',"
                + "'properties.bootstrap.servers' = 'hadoop1:9092,hadoop2:9092,hadoop3:9092',"
//                + "'format' = 'json'"
                + "'format' = 'csv'"
                + ")");

        // 使用 sql，从 kafkasource查询数据， 插入 kafkasink表
        tableEnv.executeSql("insert into sink_sensor select * from source_sensor where id = 'sensor_1'");


//        env.execute();
    }
}
/*
    使用 sql 关联外部系统：
        语法 ： create table 表名 （字段名 字段类型，字段名 字段类型.....）  with （ 参数名=参数值，参数名=参数值.....）
                   注意 ： 去官网查看 参数名 有哪些， 有一些是必需的，有一些是可选的
 */